package ihuoui.mq.store;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class CommitLog {

	// final int fileSize = 512;
	final int fileSize = 1024 * 1024 * 1024;
	// write mess
	ReentrantReadWriteLock writeReadWriteLock = new ReentrantReadWriteLock();
	ConcurrentHashMap<String, MappedFile> writeMappedFile = new ConcurrentHashMap<String, MappedFile>();
	ConcurrentHashMap<String, AtomicLong> writeOffset = new ConcurrentHashMap<String, AtomicLong>();
	// read mess
	ReentrantReadWriteLock readReadWriteLock = new ReentrantReadWriteLock();
	ConcurrentHashMap<String, MappedFile> readMappedFile = new ConcurrentHashMap<String, MappedFile>();
	ConcurrentHashMap<String, AtomicLong> readOffset = new ConcurrentHashMap<String, AtomicLong>();

	public void appendMessage(String path, Mess mess) throws StoreException {
		// if (readOffset.get(mess.getTopic()) != null &&
		// writeOffset.get(mess.getTopic()) != null) {
		// System.out.println("read:" + readOffset.get(mess.getTopic()).get() +
		// " write:"
		// + writeOffset.get(mess.getTopic()).get());
		// }
		while (true) {
			try {
				if (writeMappedFile.get(mess.getTopic()) == null) {
					writeReadWriteLock.writeLock().lock();
					try {
						writeMappedFile.putIfAbsent(mess.getTopic(), creatFirstMappedFile(path, mess.getTopic()));
					} finally {
						writeReadWriteLock.writeLock().unlock();
					}
				}

				MappedFile last = writeMappedFile.get(mess.getTopic());
				synchronized (last) {
					// System.out.println("write offset:" +
					// writeOffset.get(mess.getTopic()).get() + "\r\n");

					int ret = writeMappedFile.get(mess.getTopic()).appendMessage(ObjectToByte(mess));
					if (ret != -1) {
						// System.out.println("ret:" + ret);
						writeOffset.get(mess.getTopic()).addAndGet(ret);

						MappedFile currWMF = writeMappedFile.get(mess.getTopic());
						Long currRMLOffSet = Long.valueOf(currWMF.getFileName()
								.substring(currWMF.getFileName().length() - 20, currWMF.getFileName().length()));
						if (writeOffset.get(mess.getTopic()).get() > ((currRMLOffSet / fileSize + 1) * fileSize)) {

							MappedFile mappedFile = new MappedFile(
									path + mess.getTopic() + "/"
											+ String.format("%020d",
													(writeOffset.get(mess.getTopic()).get() / fileSize) * fileSize),
									fileSize);
							writeMappedFile.put(mess.getTopic(), mappedFile);

							// System.out.println("write create1:" +
							// String.format("%020d",
							// (writeOffset.get(mess.getTopic()).get() /
							// fileSize) * fileSize));
						}
						return;
					} else {
						// MappedFile last =
						// writeMappedFile.get(mess.getTopic());
						// synchronized (last) {
						MappedFile mappedFile = new MappedFile(
								path + mess.getTopic() + "/"
										+ String.format("%020d",
												(writeOffset.get(mess.getTopic()).get() / fileSize + 1) * fileSize),
								fileSize);
						writeMappedFile.put(mess.getTopic(), mappedFile);

						// System.out.println("write create2:" +
						// String.format("%020d",
						// (writeOffset.get(mess.getTopic()).get() / fileSize +
						// 1) * fileSize));
						// }
					}
				}
			} catch (IOException e) {
				throw new StoreException("ObjectToByte Error!");
			}

		}
	}

	public Mess offerMessage(String path, long offset, String topic) throws StoreException {

			// if (readOffset.get(topic) != null && writeOffset.get(topic) !=
			// null)
			// {
			// System.out.println("read:" + readOffset.get(topic).get() + "
			// write:"
			// + writeOffset.get(topic).get());
			// }
			if (offset > writeOffset.get(topic).get()) {
				return null;
			}
			long position = offset % fileSize;
			try {
				if (switchReadMappedFile(path, offset, topic)) {
					position = 0;
				}

				// System.out.println("offer pos:"+position+"
				// file:"+readMappedFile.get(topic).getFileName());
					MappedFileRead read = readMappedFile.get(topic)
							.getMessByPosition(new Integer(String.valueOf(position)));
					// if (read.isFlagCode() == false) {
					// synchronized (topic) {
					// if (switchReadMappedFile(path, nextFileOffSet(offset, topic),
					// topic)) {
					// position = 0;
					// read = readMappedFile.get(topic).getMessByPosition((new
					// Integer(String.valueOf(position))));
					// }
					// }
					// }
					// int curr = read.getReadPosition();
					if (read.isFlagCode()) {

						this.readOffset.get(topic).addAndGet(read.getReadLen());
						return (Mess) ByteToObject(read.getReadData());
					} else {
						return null;
					} 
			} catch (IOException | ClassNotFoundException e) {
				throw new StoreException("ObjectToByte Error!", e);
			}
	}

	private boolean switchReadMappedFile(String path, long offset, String topic) throws IOException {
		if (offset > writeOffset.get(topic).get()) {
			return false;
		}
		MappedFile currRMF = readMappedFile.get(topic);
		if (currRMF != null) {
			Long currRMLOffSet = Long.valueOf(currRMF.getFileName().substring(currRMF.getFileName().length() - 20,
					currRMF.getFileName().length()));
			if (offset < (currRMLOffSet + fileSize)) {
				return false;
			} else {
				readReadWriteLock.writeLock().lock();
				try {
					// System.out.println("switch offset:" + offset + "
					// currRMLOffSet:" + currRMLOffSet);
					readMappedFile.put(topic, creatOffSetMappedFile(path, topic, (offset / fileSize) * fileSize));
					readOffset.put(topic, new AtomicLong(offset));
				} finally {
					readReadWriteLock.writeLock().unlock();
				}
				return true;
			}
		} else {
			// System.out.println("switch offset:" + offset);
			readReadWriteLock.writeLock().lock();
			try {
				readMappedFile.put(topic, creatOffSetMappedFile(path, topic, (offset / fileSize) * fileSize));
				readOffset.put(topic, new AtomicLong(offset));
			} finally {
				readReadWriteLock.writeLock().unlock();
			}
			return true;
		}
	}

	public Mess offerNextMessage(String path, String topic) throws StoreException {
		if (this.readOffset.get(topic) == null) {
			readReadWriteLock.writeLock().lock();
			try {
				this.readOffset.putIfAbsent(topic, new AtomicLong(0l));
			} finally {
				readReadWriteLock.writeLock().unlock();
			}
		}
		Mess ret=null;
		synchronized (readOffset.get(topic)) {
			 ret = offerMessage(path, this.readOffset.get(topic).get(), topic);
		}
		return ret;
	}

	private MappedFile creatOffSetMappedFile(String path, String topic, Long fileName) throws IOException {
		return new MappedFile(path + topic + "/" + String.format("%020d", fileName), fileSize);
	}

	private MappedFile creatFirstMappedFile(String path, String topic) throws IOException {
		writeOffset.put(topic, new AtomicLong(0));
		// System.out.println("write:" + writeOffset.get(topic).get());
		// System.out
		// .println("create:" + String.format("%020d",
		// (writeOffset.get(topic).get() / fileSize + 1) * fileSize));
		return new MappedFile(path + topic + "/" + String.format("%020d", writeOffset.get(topic).get()), fileSize);
	}

	public static byte[] ObjectToByte(java.lang.Object obj) throws IOException {
		byte[] bytes = null;
		ByteArrayOutputStream bo = new ByteArrayOutputStream();
		ObjectOutputStream oo = new ObjectOutputStream(bo);
		oo.writeObject(obj);
		bytes = bo.toByteArray();
		bo.close();
		oo.close();
		return bytes;
	}

	public static Object ByteToObject(byte[] bytes) throws ClassNotFoundException, IOException {
		Object obj = null;
		// bytearray to object
		ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
		ObjectInputStream oi = new ObjectInputStream(bi);

		obj = oi.readObject();
		bi.close();
		oi.close();
		return obj;
	}

}
